/*
 * Created by Jaren at 2021/8/24 14:20
 */
import 'dart:async';
import 'dart:typed_data';

import 'package:inmotion_protocol/codec_holder.dart';
import 'package:inmotion_protocol/base/base_message.dart';
import 'package:inmotion_protocol/handler/received_handler.dart';
import 'package:inmotion_protocol/handler/send_handler.dart';
import 'package:inmotion_protocol/lorin/protocol.dart';
import 'package:inmotion_protocol/worker/worker_context.dart';
import 'package:inmotion_protocol/util/log.dart';

const int _bleTimeoutMs = 300;

const int _httpTimeoutMs = 15000;

class ProtocolWorker<T extends IMessage> implements ReceivedHandler {
  /// IMessage(waiting for response) identifier to WorkerContext instance map.
  final Map<int, WorkerContext<T>> _contexts = {};

  // mqqt队列
  final List<Map<int, Function>> mqqtFunctions = [];
  //
  bool _mqqtRequesting = false;

  /// SendHandler instance
  final SendHandler _sendHandler;

  /// unhandled message handler
  /// - param msg
  /// - return bool
  Function? unhandledMsgHandler;

  /// Once the interruptor is set, all received data will be handed over to the interruptor.
  ReceivedHandler? _dataReceiverInterruptor;

  ProtocolWorker(this._sendHandler);

  int _defaultTimeoutMs =
      ProtocolGlobal.isBleSend ? _bleTimeoutMs : _httpTimeoutMs;

  set defaultTimeoutMs(int value) {
    _defaultTimeoutMs = value;
  }

  bool get _isBleSend => ProtocolGlobal.isBleSend;

  Future<void> send(
    T msg, {
    String clientId = '',
  }) async {
    Uint8ClampedList bytes = CodecHolder.instance.encode(msg, clientId);
    if (_isBleSend) {
      _sendHandler.send(bytes, clientId);
    } else {
      _sendHandler.post(bytes, null, (rsp) {}, (errorMsg) {});
    }
  }

  Future<T> sendAndGetRspImpl(
    T msg, {
    String clientId = '',
    int? timeoutMs,
    Uint8ClampedList? encodedMsgBytes,
    int? msgId,
  }) async {
    encodedMsgBytes ??= CodecHolder.instance.encode(msg, clientId);

    msgId ??= (msg.genMsgIdentity() + clientId).hashCode;

    StreamController<T> controller = StreamController();
    if (_isBleSend) {
      WorkerContext<T> context = WorkerContext(msg);
      _contexts[msgId] = context;
      _sendHandler.send(encodedMsgBytes, clientId);
      context.timeoutTimer =
          Timer(Duration(milliseconds: timeoutMs ?? _defaultTimeoutMs), () {
        controller.sink.addError(MsgTimeoutException(msg));
        controller.close();
        // 已超时，移除对应的worker
        _contexts.remove(msgId);
      });

      context.notifier = controller;
    } else {
      WorkerContext<T> context = WorkerContext(msg);
      _contexts[msgId] = context;
      _sendHandler.post(
        encodedMsgBytes,
        msgId,
        (rsp) {
          onDataReceived(rsp);
        },
        (e) {
          if (e is MsgTimeoutException) {
            controller.sink.addError(e);
          } else {
            controller.sink.addError(MsgTimeoutException(e));
          }
          controller.close();
          // 已超时，移除对应的worker
          _contexts.remove(msgId);
        },
        clientId,
      );
      Timer(Duration(milliseconds: timeoutMs ?? _defaultTimeoutMs), () {
        if (!controller.isClosed) {
          controller.sink.addError(MsgTimeoutException(msg));
          controller.close();
          // 已超时，移除对应的worker
          _contexts.remove(msgId);
        }
      });
      context.notifier = controller;
    }
    return controller.stream.first;
  }

  Future<T> sendAndGetRsp(
    T msg, {
    String clientId = '',
    int? timeoutMs,
    int maxRetryTimes = 3,
  }) async {
    Uint8ClampedList bytes = CodecHolder.instance.encode(msg, clientId);
    int msgId = (msg.genMsgIdentity() + clientId).hashCode;
    if (_isBleSend) {
      StreamController<T> controller = StreamController();
      innerSendImpl(retriedTimes) {
        sendAndGetRspImpl(msg,
                clientId: clientId,
                timeoutMs: timeoutMs,
                encodedMsgBytes: bytes,
                msgId: msgId)
            .then((rspMsg) {
          controller.add(rspMsg);
          controller.close();
        }).catchError((e) {
          if (e is MsgTimeoutException) {
            if (retriedTimes >= maxRetryTimes) {
              /// 超过最大重试次数
              controller.addError(e);
              controller.close();
              return;
            }

            /// 重试
            innerSendImpl(retriedTimes + 1);
            print("🛸🛸🛸🛸🛸🛸 BLE 重试 ${retriedTimes + 1} ");
          }
        });
      }

      innerSendImpl(0);
      return controller.stream.first;
    } else {
      StreamController<T> controller = StreamController();
      innerSendImpl(retriedTimes) {
        sendAndGetRspImpl(msg,
                clientId: clientId,
                timeoutMs: _httpTimeoutMs,
                encodedMsgBytes: bytes,
                msgId: msgId)
            .then((rspMsg) {
          controller.add(rspMsg);
          controller.close();
        }).catchError((e) {
          controller.addError(e);
          controller.close();
        });
      }

      innerSendImpl(0);
      return controller.stream.first;
      /*
      run() {
        innerSendImpl(retriedTimes) {
          print("✅ing 执行事件");
          _mqqtRequesting = true;
          sendAndGetRspImpl(
            msg,
            clientId: clientId,
            timeoutMs: _httpTimeoutMs,
            encodedMsgBytes: bytes,
            msgId: msgId,
          ).then((rspMsg) async {
            _mqqtRequesting = false;
            controller.add(rspMsg);
            controller.close();
            mqqtFunctions.removeAt(0);
            print("✅ done 完成事件并移除，已有事件个数为：${mqqtFunctions.length}");
            if (mqqtFunctions.isNotEmpty) {
              await Future.delayed(const Duration(milliseconds: 500));
              mqqtFunctions.first.values.first();
            }
          }).catchError((e) async {
            if (e is MsgTimeoutException) {
              if (retriedTimes >= 2) {
                if (!controller.isClosed) {
                  /// 超过最大重试次数
                  _mqqtRequesting = false;
                  controller.addError(e);
                  controller.close();
                  mqqtFunctions.removeAt(0);

                  print("✅ 重试事件 timeout，已有事件个数为：${mqqtFunctions.length}");
                  if (mqqtFunctions.isNotEmpty) {
                    await Future.delayed(const Duration(milliseconds: 500));
                    mqqtFunctions.first.values.first();
                  }
                }
              } else {
                /// 重试
                await Future.delayed(const Duration(milliseconds: 1000));
                innerSendImpl(retriedTimes + 1);
                print("❌ 重试事件 = ${retriedTimes + 1}");
              }
            } else {
              if (!controller.isClosed) {
                _mqqtRequesting = false;
                controller.addError(e);
                controller.close();
                mqqtFunctions.removeAt(0);

                print("✅ 时间错误，已有事件个数为：${mqqtFunctions.length}");
                if (mqqtFunctions.isNotEmpty) {
                  mqqtFunctions.first.values.first();
                }
              }
            }
          });
        }

        innerSendImpl(0);
      }

      Map<int, Function> map = Map();
      map[msgId] = run;
      mqqtFunctions.add(map);
      if (mqqtFunctions.length == 1 && !_mqqtRequesting) {
        print("✅ 执行第一条事件");
        await Future.delayed(const Duration(milliseconds: 500));
        mqqtFunctions.first.values.first();
      }
        */
      return controller.stream.first;
    }
  }

  Future<T> getRspMsg(T msg) {
    return Future(() => msg);
  }

  @override
  void onDataReceived(
    Uint8ClampedList bytes, [
    String clientId = '',
  ]) {
    if (_dataReceiverInterruptor != null) {
      _dataReceiverInterruptor!.onDataReceived(bytes, clientId);
      return;
    }
    List<T> receivedMsgList = CodecHolder.instance.decode(bytes, []);
    if (receivedMsgList.isEmpty) return;
    while (receivedMsgList.isNotEmpty) {
      T msg = receivedMsgList.removeAt(0);

      int msgId = _msgId4WorkerMap(msg, clientId);
      LogUtils.v(
          'onDataReceived: Received msg with id [$msgId] for client [$clientId].');
      WorkerContext<T>? context = _contexts[msgId];
      if (context == null) {
        if (unhandledMsgHandler == null) continue;
        if (!unhandledMsgHandler!(msg) && _isBleSend) {
          LogUtils.v(
              'onDataReceived: Unhandled received msg. There is no msg with client [$clientId] waiting for rsp.');
        }
        continue;
      }

      // 通知收到回复数据
      context.rspMsg = msg;
      if (context.notifier != null) {
        context.notifier!.add(msg);
        context.notifier!.close();
        // 已收到回复，移除对应的worker
        LogUtils.v(
            'onDataReceived: Notified the waiting msg with id [$msgId] for client [$clientId].');
        _contexts.remove(msgId);
      }
      if (context.timeoutTimer != null) context.timeoutTimer!.cancel();
    }
  }

  int _msgId4WorkerMap(T msg, String clientId) {
    return (msg.genMsgIdentity() + clientId).hashCode;
  }

  set dataReceiverInterruptor(ReceivedHandler? value) {
    _dataReceiverInterruptor = value;
  }

  SendHandler get sendHandler => _sendHandler;
}

class MsgTimeoutException<T> implements Exception {
  T msg;

  MsgTimeoutException(this.msg);

  @override
  String toString() {
    return "Message timeout, source msg: " + msg.toString();
  }
}

class UnknownProtocolWorker extends ProtocolWorker {
  static final ProtocolWorker instance =
      UnknownProtocolWorker(UnknownSendHandler());

  UnknownProtocolWorker(SendHandler sendHandler) : super(sendHandler);
}
